package com.sunday.common.mq.rocket.brave.study.apache.e5_Request_Reply_Message_Exchange_Pattern;

import com.sunday.common.mq.rocket.brave.study.apache.User;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQLocalRequestCallback;
import org.apache.rocketmq.spring.core.RocketMQReplyListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * 请求回复消息交换模式
 * <p>
 * RocketMQ-Spring 提供请求/回复语义支持。
 * 使用 sendAndReceice 方法发送请求消息。
 * <p>
 * 注意：
 * 同步发送需要在方法的参数中指示返回值类型。
 * 异步发送需要在回调接口中指示返回值类型。
 * <p>
 * https://github.com/apache/rocketmq-spring/wiki/Request-reply-message-exchange-pattern
 */
@Slf4j
@RestController
@RequestMapping("/test")
public class SendAndReceiceApi {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    private static final String topic = "your_topic";

    @GetMapping("send")
    public void send() {
        User user = User.builder().id(1).userName("john" + 1).userAge(33 + 1).build();
        Message<User> message = MessageBuilder.withPayload(user).build();
        // 以同步模式发送请求并接收字符串类型的回复。
        String replyString = rocketMQTemplate.sendAndReceive(topic, message, String.class);
        log.info("sync : {}", replyString);
    }

    @GetMapping("asyncSend")
    public void asyncSend() {
        User user = User.builder().id(2).userName("john" + 2).userAge(33 + 2).build();
        Message<User> message = MessageBuilder.withPayload(user).build();
        // 以异步方式发送请求并接收用户类型的回复。
        rocketMQTemplate.sendAndReceive(topic, message, new RocketMQLocalRequestCallback<MessageExt>() {
            @Override
            public void onSuccess(MessageExt message) {
                log.info("async : {}", new String(message.getBody()));
            }

            @Override
            public void onException(Throwable e) {
                e.printStackTrace();
            }
        }, 20000);
    }

    @Slf4j
    @Service
    @RocketMQMessageListener(
            topic = "your_topic",
            consumerGroup = "your_consumer_group_name"
//            , consumeThreadNumber = 1
    )
    public static class MessageExtConsumer implements RocketMQReplyListener<MessageExt, String> {
        @Override
        public String onMessage(MessageExt message) {
            log.info("{} - {}", new String(message.getBody()), message);
            return "reply string";
        }
    }

}